package memberlist

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"os"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/hashicorp/memberlist"
	"github.com/louloulin/dataflare/pkg/cluster"
)

// Manager 是基于 Memberlist 的集群管理器
type Manager struct {
	// config 是集群配置
	config *cluster.Config
	// memberlist 是成员列表
	memberlist *memberlist.Memberlist
	// delegate 是成员列表委托
	delegate *delegate
	// nodes 是节点信息映射
	nodes map[string]*cluster.NodeInfo
	// localNode 是本地节点信息
	localNode *cluster.NodeInfo
	// eventChannels 是事件通道映射
	eventChannels map[string][]chan cluster.Event
	// mu 是保护节点和事件通道的互斥锁
	mu sync.RWMutex
	// logger 是日志记录器
	logger *log.Logger
	// ctx 是上下文
	ctx context.Context
	// cancel 是取消函数
	cancel context.CancelFunc
}

// delegate 是成员列表委托
type delegate struct {
	// manager 是集群管理器
	manager *Manager
}

// NodeMeta 实现 memberlist.Delegate 接口
func (d *delegate) NodeMeta(limit int) []byte {
	d.manager.mu.RLock()
	defer d.manager.mu.RUnlock()

	// 序列化节点元数据
	meta := map[string]string{
		"role":    d.manager.localNode.Role,
		"version": d.manager.localNode.Version,
	}

	// 添加标签
	for i, tag := range d.manager.localNode.Tags {
		meta[fmt.Sprintf("tag_%d", i)] = tag
	}

	// 添加自定义元数据
	for k, v := range d.manager.localNode.Metadata {
		meta[k] = v
	}

	// 序列化为 JSON
	data, err := json.Marshal(meta)
	if err != nil {
		d.manager.logger.Printf("Failed to marshal node meta: %v", err)
		return []byte{}
	}

	// 限制大小
	if len(data) > limit {
		d.manager.logger.Printf("Node meta exceeds limit: %d > %d", len(data), limit)
		return data[:limit]
	}

	return data
}

// NotifyMsg 实现 memberlist.Delegate 接口
func (d *delegate) NotifyMsg(msg []byte) {
	if len(msg) == 0 {
		return
	}

	// 解析消息
	var message struct {
		Type string          `json:"type"`
		Data json.RawMessage `json:"data"`
	}

	if err := json.Unmarshal(msg, &message); err != nil {
		d.manager.logger.Printf("Failed to unmarshal message: %v", err)
		return
	}

	// 处理消息
	switch message.Type {
	case "broadcast":
		d.manager.handleBroadcast(message.Data)
	case "metadata_update":
		d.manager.handleMetadataUpdate(message.Data)
	default:
		d.manager.logger.Printf("Unknown message type: %s", message.Type)
	}
}

// GetBroadcasts 实现 memberlist.Delegate 接口
func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
	return nil
}

// LocalState 实现 memberlist.Delegate 接口
func (d *delegate) LocalState(join bool) []byte {
	return nil
}

// MergeRemoteState 实现 memberlist.Delegate 接口
func (d *delegate) MergeRemoteState(buf []byte, join bool) {
}

// eventDelegate 是成员列表事件委托
type eventDelegate struct {
	// manager 是集群管理器
	manager *Manager
}

// NotifyJoin 实现 memberlist.EventDelegate 接口
func (e *eventDelegate) NotifyJoin(node *memberlist.Node) {
	e.manager.handleNodeJoin(node)
}

// NotifyLeave 实现 memberlist.EventDelegate 接口
func (e *eventDelegate) NotifyLeave(node *memberlist.Node) {
	e.manager.handleNodeLeave(node)
}

// NotifyUpdate 实现 memberlist.EventDelegate 接口
func (e *eventDelegate) NotifyUpdate(node *memberlist.Node) {
	e.manager.handleNodeUpdate(node)
}

// NewManager 创建一个新的基于 Memberlist 的集群管理器
func NewManager(config *cluster.Config) (cluster.Manager, error) {
	// 验证配置
	if err := config.Validate(); err != nil {
		return nil, err
	}

	// 生成节点ID
	if config.NodeID == "" {
		config.NodeID = uuid.New().String()
	}

	// 创建上下文
	ctx, cancel := context.WithCancel(context.Background())

	// 创建管理器
	manager := &Manager{
		config:        config,
		nodes:         make(map[string]*cluster.NodeInfo),
		eventChannels: make(map[string][]chan cluster.Event),
		logger:        log.New(os.Stderr, "[MemberlistManager] ", log.LstdFlags),
		ctx:           ctx,
		cancel:        cancel,
	}

	// 创建委托
	delegate := &delegate{
		manager: manager,
	}

	// 创建事件委托
	eventDelegate := &eventDelegate{
		manager: manager,
	}

	// 解析绑定地址
	bindHost, bindPort, err := cluster.ParseAddress(config.BindAddress)
	if err != nil {
		return nil, fmt.Errorf("invalid bind address: %w", err)
	}

	// 创建 Memberlist 配置
	mlConfig := memberlist.DefaultLANConfig()
	mlConfig.Name = config.NodeID
	mlConfig.BindAddr = bindHost
	mlConfig.BindPort = bindPort
	mlConfig.Delegate = delegate
	mlConfig.Events = eventDelegate
	mlConfig.LogOutput = os.Stderr

	// 设置广播地址
	if config.AdvertiseAddress != "" {
		advHost, advPort, err := cluster.ParseAddress(config.AdvertiseAddress)
		if err != nil {
			return nil, fmt.Errorf("invalid advertise address: %w", err)
		}
		mlConfig.AdvertiseAddr = advHost
		mlConfig.AdvertisePort = advPort
	}

	// 创建 Memberlist
	ml, err := memberlist.Create(mlConfig)
	if err != nil {
		return nil, fmt.Errorf("failed to create memberlist: %w", err)
	}

	// 设置 Memberlist
	manager.memberlist = ml
	manager.delegate = delegate

	// 创建本地节点信息
	localIP := ml.LocalNode().Addr.String()
	localPort := ml.LocalNode().Port
	manager.localNode = &cluster.NodeInfo{
		ID:       config.NodeID,
		Address:  cluster.FormatAddress(localIP, localPort),
		Status:   cluster.NodeStatusUp,
		LastSeen: time.Now(),
		Metadata: config.Metadata,
		Role:     config.Role,
		Tags:     config.Tags,
		Version:  "1.0.0",
	}

	// 注册本地节点
	manager.nodes[config.NodeID] = manager.localNode

	return manager, nil
}

// Start 启动集群管理器
func (m *Manager) Start(ctx context.Context) error {
	m.logger.Println("Starting cluster manager")

	// 加入种子节点
	if len(m.config.SeedAddresses) > 0 {
		_, err := m.memberlist.Join(m.config.SeedAddresses)
		if err != nil {
			m.logger.Printf("Failed to join seed nodes: %v", err)
			// 继续运行，不返回错误
		}
	}

	// 启动定期任务
	go m.periodicTasks()

	return nil
}

// Stop 停止集群管理器
func (m *Manager) Stop(ctx context.Context) error {
	m.logger.Println("Stopping cluster manager")

	// 取消上下文
	m.cancel()

	// 离开集群
	err := m.memberlist.Leave(m.config.LeaveTimeout)
	if err != nil {
		m.logger.Printf("Failed to leave cluster: %v", err)
	}

	// 关闭 Memberlist
	err = m.memberlist.Shutdown()
	if err != nil {
		m.logger.Printf("Failed to shutdown memberlist: %v", err)
	}

	// 关闭所有事件通道
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, channels := range m.eventChannels {
		for _, ch := range channels {
			close(ch)
		}
	}
	m.eventChannels = make(map[string][]chan cluster.Event)

	return nil
}

// Join 加入集群
func (m *Manager) Join(ctx context.Context, addresses []string) error {
	m.logger.Printf("Joining cluster with addresses: %v", addresses)

	// 加入集群
	n, err := m.memberlist.Join(addresses)
	if err != nil {
		return fmt.Errorf("failed to join cluster: %w", err)
	}

	m.logger.Printf("Joined cluster with %d nodes", n)
	return nil
}

// Leave 离开集群
func (m *Manager) Leave(ctx context.Context) error {
	m.logger.Println("Leaving cluster")

	// 离开集群
	err := m.memberlist.Leave(m.config.LeaveTimeout)
	if err != nil {
		return fmt.Errorf("failed to leave cluster: %w", err)
	}

	return nil
}

// GetNodes 获取集群节点
func (m *Manager) GetNodes(ctx context.Context) ([]*cluster.NodeInfo, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// 创建节点列表
	nodes := make([]*cluster.NodeInfo, 0, len(m.nodes))
	for _, node := range m.nodes {
		nodes = append(nodes, node)
	}

	return nodes, nil
}

// GetNode 获取指定节点
func (m *Manager) GetNode(ctx context.Context, nodeID string) (*cluster.NodeInfo, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// 查找节点
	node, ok := m.nodes[nodeID]
	if !ok {
		return nil, fmt.Errorf("node not found: %s", nodeID)
	}

	return node, nil
}

// GetLocalNode 获取本地节点
func (m *Manager) GetLocalNode(ctx context.Context) (*cluster.NodeInfo, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	return m.localNode, nil
}

// IsLeader 检查本地节点是否为领导者
func (m *Manager) IsLeader(ctx context.Context) (bool, error) {
	leader, err := m.GetLeader(ctx)
	if err != nil {
		return false, err
	}

	return leader.ID == m.localNode.ID, nil
}

// GetLeader 获取领导者节点
func (m *Manager) GetLeader(ctx context.Context) (*cluster.NodeInfo, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// 获取所有节点
	members := m.memberlist.Members()
	if len(members) == 0 {
		return nil, fmt.Errorf("no nodes in cluster")
	}

	// 选择第一个节点作为领导者
	leaderNode := members[0]
	leader, ok := m.nodes[leaderNode.Name]
	if !ok {
		return nil, fmt.Errorf("leader node not found: %s", leaderNode.Name)
	}

	return leader, nil
}

// Subscribe 订阅集群事件
func (m *Manager) Subscribe(ctx context.Context, eventType string) (<-chan cluster.Event, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// 创建事件通道
	ch := make(chan cluster.Event, 10)

	// 添加到事件通道映射
	m.eventChannels[eventType] = append(m.eventChannels[eventType], ch)

	return ch, nil
}

// Unsubscribe 取消订阅集群事件
func (m *Manager) Unsubscribe(ctx context.Context, eventType string, ch <-chan cluster.Event) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// 查找事件通道
	channels, ok := m.eventChannels[eventType]
	if !ok {
		return fmt.Errorf("no channels for event type: %s", eventType)
	}

	// 移除事件通道
	for i, c := range channels {
		if c == ch {
			close(c)
			m.eventChannels[eventType] = append(channels[:i], channels[i+1:]...)
			return nil
		}
	}

	return fmt.Errorf("channel not found for event type: %s", eventType)
}

// Broadcast 广播消息到集群
func (m *Manager) Broadcast(ctx context.Context, message []byte) error {
	// 创建广播消息
	broadcastMsg := struct {
		Type string          `json:"type"`
		Data json.RawMessage `json:"data"`
	}{
		Type: "broadcast",
		Data: message,
	}

	// 序列化消息
	data, err := json.Marshal(broadcastMsg)
	if err != nil {
		return fmt.Errorf("failed to marshal broadcast message: %w", err)
	}

	// 广播消息
	err = m.memberlist.SendBestEffort(m.memberlist.LocalNode(), data)
	if err != nil {
		return fmt.Errorf("failed to broadcast message: %w", err)
	}

	return nil
}

// Send 发送消息到指定节点
func (m *Manager) Send(ctx context.Context, nodeID string, message []byte) error {
	// 查找节点
	var targetNode *memberlist.Node
	for _, node := range m.memberlist.Members() {
		if node.Name == nodeID {
			targetNode = node
			break
		}
	}

	if targetNode == nil {
		return fmt.Errorf("node not found: %s", nodeID)
	}

	// 创建广播消息
	broadcastMsg := struct {
		Type string          `json:"type"`
		Data json.RawMessage `json:"data"`
	}{
		Type: "broadcast",
		Data: message,
	}

	// 序列化消息
	data, err := json.Marshal(broadcastMsg)
	if err != nil {
		return fmt.Errorf("failed to marshal broadcast message: %w", err)
	}

	// 发送消息
	err = m.memberlist.SendBestEffort(targetNode, data)
	if err != nil {
		return fmt.Errorf("failed to send message: %w", err)
	}

	return nil
}

// SetMetadata 设置节点元数据
func (m *Manager) SetMetadata(ctx context.Context, key string, value string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// 更新本地节点元数据
	m.localNode.Metadata[key] = value

	// 创建元数据更新消息
	metadataMsg := struct {
		Type string            `json:"type"`
		Data map[string]string `json:"data"`
	}{
		Type: "metadata_update",
		Data: map[string]string{
			"node_id": m.localNode.ID,
			"key":     key,
			"value":   value,
		},
	}

	// 序列化消息
	data, err := json.Marshal(metadataMsg)
	if err != nil {
		return fmt.Errorf("failed to marshal metadata message: %w", err)
	}

	// 广播消息
	err = m.memberlist.SendBestEffort(m.memberlist.LocalNode(), data)
	if err != nil {
		return fmt.Errorf("failed to broadcast metadata update: %w", err)
	}

	return nil
}

// GetMetadata 获取节点元数据
func (m *Manager) GetMetadata(ctx context.Context, nodeID string, key string) (string, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// 查找节点
	node, ok := m.nodes[nodeID]
	if !ok {
		return "", fmt.Errorf("node not found: %s", nodeID)
	}

	// 获取元数据
	value, ok := node.Metadata[key]
	if !ok {
		return "", fmt.Errorf("metadata key not found: %s", key)
	}

	return value, nil
}

// handleNodeJoin 处理节点加入事件
func (m *Manager) handleNodeJoin(node *memberlist.Node) {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.logger.Printf("Node joined: %s at %s", node.Name, node.Address())

	// 解析元数据
	metadata := make(map[string]string)
	if len(node.Meta) > 0 {
		if err := json.Unmarshal(node.Meta, &metadata); err != nil {
			m.logger.Printf("Failed to unmarshal node meta: %v", err)
		}
	}

	// 获取角色和标签
	role := metadata["role"]
	delete(metadata, "role")

	version := metadata["version"]
	delete(metadata, "version")

	var tags []string
	for k, v := range metadata {
		if len(k) > 4 && k[:4] == "tag_" {
			tags = append(tags, v)
			delete(metadata, k)
		}
	}

	// 创建节点信息
	nodeInfo := &cluster.NodeInfo{
		ID:       node.Name,
		Address:  cluster.FormatAddress(node.Addr.String(), int(node.Port)),
		Status:   cluster.NodeStatusUp,
		LastSeen: time.Now(),
		Metadata: metadata,
		Role:     role,
		Tags:     tags,
		Version:  version,
	}

	// 更新节点信息
	m.nodes[node.Name] = nodeInfo

	// 发布事件
	m.publishEvent(cluster.EventTypeNodeJoin, nodeInfo)
}

// handleNodeLeave 处理节点离开事件
func (m *Manager) handleNodeLeave(node *memberlist.Node) {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.logger.Printf("Node left: %s", node.Name)

	// 查找节点
	nodeInfo, ok := m.nodes[node.Name]
	if !ok {
		m.logger.Printf("Node not found: %s", node.Name)
		return
	}

	// 更新节点状态
	nodeInfo.Status = cluster.NodeStatusLeaving
	nodeInfo.LastSeen = time.Now()

	// 发布事件
	m.publishEvent(cluster.EventTypeNodeLeave, nodeInfo)
}

// handleNodeUpdate 处理节点更新事件
func (m *Manager) handleNodeUpdate(node *memberlist.Node) {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.logger.Printf("Node updated: %s", node.Name)

	// 查找节点
	nodeInfo, ok := m.nodes[node.Name]
	if !ok {
		m.logger.Printf("Node not found: %s", node.Name)
		return
	}

	// 解析元数据
	metadata := make(map[string]string)
	if len(node.Meta) > 0 {
		if err := json.Unmarshal(node.Meta, &metadata); err != nil {
			m.logger.Printf("Failed to unmarshal node meta: %v", err)
		}
	}

	// 获取角色和标签
	role := metadata["role"]
	delete(metadata, "role")

	version := metadata["version"]
	delete(metadata, "version")

	var tags []string
	for k, v := range metadata {
		if len(k) > 4 && k[:4] == "tag_" {
			tags = append(tags, v)
			delete(metadata, k)
		}
	}

	// 更新节点信息
	nodeInfo.Address = cluster.FormatAddress(node.Addr.String(), int(node.Port))
	nodeInfo.LastSeen = time.Now()
	nodeInfo.Metadata = metadata
	nodeInfo.Role = role
	nodeInfo.Tags = tags
	nodeInfo.Version = version

	// 发布事件
	m.publishEvent(cluster.EventTypeNodeUpdate, nodeInfo)
}

// handleBroadcast 处理广播消息
func (m *Manager) handleBroadcast(data json.RawMessage) {
	// 发布事件
	event := cluster.Event{
		Type:      "broadcast",
		Timestamp: time.Now(),
		Data: map[string]interface{}{
			"message": data,
		},
	}

	m.mu.RLock()
	defer m.mu.RUnlock()

	// 发送到所有广播事件通道
	for _, ch := range m.eventChannels["broadcast"] {
		select {
		case ch <- event:
		default:
			m.logger.Println("Broadcast event channel full, dropping event")
		}
	}
}

// handleMetadataUpdate 处理元数据更新消息
func (m *Manager) handleMetadataUpdate(data json.RawMessage) {
	// 解析元数据更新消息
	var metadataUpdate struct {
		NodeID string `json:"node_id"`
		Key    string `json:"key"`
		Value  string `json:"value"`
	}

	if err := json.Unmarshal(data, &metadataUpdate); err != nil {
		m.logger.Printf("Failed to unmarshal metadata update: %v", err)
		return
	}

	m.mu.Lock()
	defer m.mu.Unlock()

	// 查找节点
	nodeInfo, ok := m.nodes[metadataUpdate.NodeID]
	if !ok {
		m.logger.Printf("Node not found: %s", metadataUpdate.NodeID)
		return
	}

	// 更新元数据
	nodeInfo.Metadata[metadataUpdate.Key] = metadataUpdate.Value

	// 发布事件
	m.publishEvent(cluster.EventTypeNodeUpdate, nodeInfo)
}

// publishEvent 发布事件
func (m *Manager) publishEvent(eventType string, nodeInfo *cluster.NodeInfo) {
	// 创建事件
	event := cluster.Event{
		Type:      eventType,
		NodeID:    nodeInfo.ID,
		Node:      nodeInfo,
		Timestamp: time.Now(),
	}

	// 发送到所有事件通道
	for _, ch := range m.eventChannels[eventType] {
		select {
		case ch <- event:
		default:
			m.logger.Printf("Event channel full, dropping event: %s", eventType)
		}
	}

	// 发送到所有集群变更事件通道
	for _, ch := range m.eventChannels[cluster.EventTypeClusterChanged] {
		select {
		case ch <- event:
		default:
			m.logger.Println("Cluster changed event channel full, dropping event")
		}
	}
}

// periodicTasks 执行定期任务
func (m *Manager) periodicTasks() {
	// 创建定时器
	heartbeatTicker := time.NewTicker(m.config.HeartbeatInterval)
	defer heartbeatTicker.Stop()

	for {
		select {
		case <-m.ctx.Done():
			return
		case <-heartbeatTicker.C:
			m.checkNodes()
		}
	}
}

// checkNodes 检查节点状态
func (m *Manager) checkNodes() {
	m.mu.Lock()
	defer m.mu.Unlock()

	// 获取当前时间
	now := time.Now()

	// 检查所有节点
	for id, node := range m.nodes {
		// 跳过本地节点
		if id == m.localNode.ID {
			continue
		}

		// 检查节点是否超时
		if node.Status == cluster.NodeStatusUp && now.Sub(node.LastSeen) > m.config.HeartbeatTimeout {
			// 更新节点状态
			node.Status = cluster.NodeStatusDown
			m.logger.Printf("Node down: %s", id)

			// 发布事件
			m.publishEvent(cluster.EventTypeNodeFailed, node)
		}
	}
}

// init 注册集群管理器工厂
func init() {
	cluster.RegisterManagerFactory("memberlist", func(config *cluster.Config) (cluster.Manager, error) {
		return NewManager(config)
	})
}
